using System;
using System.Collections.Generic;
using System.Text;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Service.Template;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
    /// <summary>
    /// Iterates over the range slices, automatically refreshing the query until all matching rows are returned.
    /// Note: we use PeekEnumerator to iterate over the resultset, since IEnumerator doesn't have an easy way
    /// to peek the next element.
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
    /// <typeparam name="V"></typeparam>
    public class RangeSlicesEnumerator<K, N, V> : IEnumerator<Row<K, N, V>>
	{
        private RangeSlicesQuery<K, N, V> query;
        private PeekEnumerator<K, N, V> enumerator;

        public RangeSlicesEnumerator(RangeSlicesQuery<K, N, V> query, K startKey, K endKey)
        {
            this.query = query;
            StartKey = startKey;
            EndKey = endKey;
            this.query.SetKeys(startKey, endKey);
            Result = this.query.Execute();
            Rows = Result.Get().Count;
            enumerator = new PeekEnumerator<K, N, V>(Result.Get().List);
        }
        private K StartKey
        {
            get;
            set;
        }
        private K EndKey
        {
            get;
            set;
        }
        private int Rows
        {
            get;
            set;
        }
        private QueryResult<OrderedRows<K, N, V>> Result
        {
            get;
            set;
        }
        public SliceFilter<Row<K, N, V>> Filter
        {
            get;
            set;
        }
        /// <summary>
        /// The current row is only used internally and isn't visible
        /// to outside classes.
        /// </summary>
        private Row<K, N, V> CurrentRow
        {
            get;
            set;
        }
        public bool MoveNext()
        {
            if (enumerator == null)
            {
                enumerator = new PeekEnumerator<K, N, V>(Result.Get().List);
            }
            else if (!enumerator.HasNext() && Rows == this.query.RowCount)
            {
                this.query.SetKeys(StartKey, EndKey);
                Result = this.query.Execute();
                enumerator = new PeekEnumerator<K, N, V>(Result.Get().List);
                Rows = 0;
                if (enumerator.HasNext())
                {
                    enumerator.MoveNext(); // we skip the first one, since it is the last record in the previous query
                }
            }
            while (Filter != null && enumerator != null && enumerator.HasNext() && !Filter.Accept(enumerator.Peek()))
            {
                MoveNext();
                if (!enumerator.HasNext() && Rows == query.RowCount)
                {
                    Refresh();
                }
            }
            return enumerator.MoveNext();
        }
        public new Row<K, N, V> Current
        {
            get
            {
                CurrentRow = enumerator.Current;
                StartKey = CurrentRow.Key;
                Rows++;
                return CurrentRow; 
            }
        }
        object System.Collections.IEnumerator.Current
        {
            get { return this.Current; }
        }
        private void Refresh()
        {
            this.query.SetKeys(StartKey, EndKey);
            Result = this.query.Execute();
            enumerator = new PeekEnumerator<K, N, V>(Result.Get().List);
            Rows = 0;

            if (enumerator.HasNext())
            {
                // First element is startKey which was the last element on the previous query result - skip it
                enumerator.MoveNext();
            }
        }
        public void Reset()
        {
            enumerator = null;
        }
        public void Dispose()
        {
            enumerator = null;
            StartKey = default(K);
            EndKey = default(K);
            Result = null;
        }
	}
}